package com.test.rxjavademo;

import android.util.Log;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

public class ToolOperatorDemo {
    public static final String TAG = "ToolOperatorDemo";

    public void test1(){
        Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
                Log.e(TAG, "subscribe: " + Thread.currentThread());

                emitter.onNext("第一个事件");
                emitter.onNext("第二个事件");
                emitter.onNext("第三个事件");
                emitter.onComplete();

            }
        }).subscribeOn(Schedulers.newThread())  //主要决定我们执行subscribe方法所处的线程，也就是产生事件发射事件所在的线程
                .observeOn(AndroidSchedulers.mainThread())  //来决定下游事件所处的线程
                .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe: " + Thread.currentThread());
            }

            @Override
            public void onNext(Object o) {
                Log.e(TAG, "onNext: " + Thread.currentThread());
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + Thread.currentThread());
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete: " + Thread.currentThread());
            }
        });
    }
}
